package io.kiki.sba.registry.store.jdbc.repository.impl;

import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.metaserver.ClientManagerAddress;
import io.kiki.sba.registry.common.model.metaserver.ClientManagerAddress.AddressVersion;
import io.kiki.sba.registry.common.model.metaserver.ClientManagerResult;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.date.DateNowRepository;
import io.kiki.sba.registry.store.api.meta.ClientManagerAddressRepository;
import io.kiki.sba.registry.store.api.meta.RecoverConfig;
import io.kiki.sba.registry.store.jdbc.config.MetadataConfig;
import io.kiki.sba.registry.store.jdbc.constant.TableEnum;
import io.kiki.sba.registry.store.jdbc.domain.ClientManagerAddressDomain;
import io.kiki.sba.registry.store.jdbc.informer.BaseInformer;
import io.kiki.sba.registry.store.jdbc.mapper.ClientManagerAddressMapper;
import io.kiki.sba.registry.util.NamedThreadFactory;
import io.kiki.sba.registry.util.StringFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;

import static io.kiki.sba.registry.store.jdbc.repository.impl.MetadataMetrics.ProvideData.CLIENT_MANAGER_UPDATE_COUNTER;


public class ClientManagerAddressJdbcRepository implements ClientManagerAddressRepository, RecoverConfig {

    private static final Logger logger = LoggerFactory.getLogger(ClientManagerAddressJdbcRepository.class);
    final Informer informer;
    @Autowired
    private DefaultCommonConfig defaultCommonConfig;
    @Autowired
    private ClientManagerAddressMapper clientManagerAddressMapper;
    @Autowired
    private DateNowRepository dateNowRepository;
    @Autowired
    private MetadataConfig metadataConfig;
    private ThreadPoolExecutor threadPoolExecutor;

    public ClientManagerAddressJdbcRepository() {
        informer = new Informer();
    }

    public ClientManagerAddressJdbcRepository(MetadataConfig metadataConfig) {
        informer = new Informer();
        this.metadataConfig = metadataConfig;
    }

    @PostConstruct
    public void init() {
        threadPoolExecutor = new ThreadPoolExecutor(metadataConfig.getClientManagerExecutorPoolSize(), metadataConfig.getClientManagerExecutorPoolSize(), 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(metadataConfig.getClientManagerExecutorQueueSize()), new NamedThreadFactory("ClientManagerExecutor"), new CallerRunsPolicy());

        informer.setEnabled(true);
        informer.start();
    }

    @Override
    public ClientManagerResult clientOpen(Set<AddressVersion> ipSet) {
        try {
            long maxId = doStorage(ipSet, ValueConstants.CLIENT_OPEN);
            CLIENT_MANAGER_UPDATE_COUNTER.inc(ipSet.size());
            return ClientManagerResult.buildSuccess(maxId);
        } catch (Throwable t) {
            logger.error("clientOpen:{} error.", ipSet, t);
            return ClientManagerResult.buildFailRet();
        }
    }

    @Override
    public ClientManagerResult clientOff(Set<AddressVersion> ipSet) {
        try {
            long maxId = doStorage(ipSet, ValueConstants.CLIENT_OFF);
            CLIENT_MANAGER_UPDATE_COUNTER.inc(ipSet.size());
            return ClientManagerResult.buildSuccess(maxId);
        } catch (Throwable t) {
            logger.error("clientOff:{} error.", ipSet, t);
            return ClientManagerResult.buildFailRet();
        }
    }

    @Override
    public ClientManagerResult reduce(Set<AddressVersion> ipSet) {
        try {
            long maxId = doStorage(ipSet, ValueConstants.REDUCE);
            CLIENT_MANAGER_UPDATE_COUNTER.inc(ipSet.size());
            return ClientManagerResult.buildSuccess(maxId);
        } catch (Throwable t) {
            logger.error("clientOff:{} error.", ipSet, t);
            return ClientManagerResult.buildFailRet();
        }
    }

    @Override
    public ClientManagerAddress queryClientOffData() {
        long version = informer.getLastLoadId();
        ClientManagerAddressContainer.ClientManagerAddress query = informer.getContainer().queryClientManagerAddress();
        return new ClientManagerAddress(version, query.getClientOffData(), query.getReduces());
    }

    @Override
    public void waitSynced() {
        informer.waitSynced();
    }

    @Override
    public List<String> getExpireAddress(Date date, int limit) {
        return clientManagerAddressMapper.getExpireAddress(defaultCommonConfig.getClusterId(tableName()), date, limit);
    }

    @Override
    public int cleanExpired(List<String> expireAddress) {
        if (CollectionUtils.isEmpty(expireAddress)) {
            return 0;
        }

        return clientManagerAddressMapper.cleanExpired(defaultCommonConfig.getClusterId(tableName()), expireAddress);
    }

    @Override
    public int getClientOffSizeBefore(Date date) {
        return clientManagerAddressMapper.getClientOffSizeBefore(defaultCommonConfig.getClusterId(tableName()), date);
    }

    @Override
    public void wakeup() {
        informer.watchWakeup();
    }

    private long doStorage(Set<AddressVersion> ipSet, String operation) throws ExecutionException, InterruptedException, TimeoutException {
        List<Future<Tuple<Boolean, Long>>> list = new ArrayList<>();
        for (AddressVersion addressVersion : ipSet) {
            Future<Tuple<Boolean, Long>> future = threadPoolExecutor.submit(() -> {
                long maxId = 0;
                ClientManagerAddressDomain clientManagerAddressDomain = new ClientManagerAddressDomain(defaultCommonConfig.getClusterId(tableName()), addressVersion.getAddress(), operation, addressVersion.isPub(), addressVersion.isSub());
                int effectRows = clientManagerAddressMapper.update(clientManagerAddressDomain);

                if (effectRows == 0) {
                    clientManagerAddressMapper.insertOnReplace(clientManagerAddressDomain);
                    logger.info("address: {} replace.", clientManagerAddressDomain);
                    maxId = clientManagerAddressDomain.getId();
                } else {
                    logger.info("address: {} exist, skip replace.", clientManagerAddressDomain);
                }
                return new Tuple<>(true, maxId);
            });
            list.add(future);
        }

        long maxId = 0;
        for (Future<Tuple<Boolean, Long>> future : list) {
            Tuple<Boolean, Long> tuple = future.get(3000, TimeUnit.MILLISECONDS);
            if (tuple.o1 == null || !tuple.o1) {
                throw new SofaRegistryRuntimeException(StringFormatter.format("ips: {}, operation:{} execute error.", ipSet, operation));
            }
            if (tuple.o2 > maxId) {
                maxId = tuple.o2;
            }
        }

        return maxId;
    }

    @Override
    public String tableName() {
        return TableEnum.client_manager_address.getTableName();
    }


    class Informer extends BaseInformer<ClientManagerAddressDomain, ClientManagerAddressContainer> {

        public Informer() {
            super("ClientManager");
        }

        @Override
        protected ClientManagerAddressContainer containerFactory() {
            return new ClientManagerAddressContainer();
        }

        @Override
        protected List<ClientManagerAddressDomain> listFromStorage(long start, int limit) {
            return clientManagerAddressMapper.queryAfterThanByLimit(defaultCommonConfig.getClusterId(tableName()), start, limit);
        }

        @Override
        protected Date getNow() {
            return dateNowRepository.getNow();
        }
    }
}
